/*-------------------------------------------------------------------------
 *
 * xloginsert.c
 *		构建 WAL 记录的函数
 *
 * 构建 WAL 记录始于调用 XLogBeginInsert，随后进行若干 XLogRegister* 调用。
 * 注册的数据存储在私有工作内存中，最后通过调用 XLogRecordAssemble() 将其汇编
 * 成一系列 XLogRecData 结构。有关详细信息，请参阅 access/transam/README。
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 * src/backend/access/transam/xloginsert.c
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#ifdef USE_LZ4
#include <lz4.h>
#endif

#ifdef USE_ZSTD
#include <zstd.h>
#endif

#include "access/xact.h"
#include "access/xlog.h"
#include "access/xlog_internal.h"
#include "access/xloginsert.h"
#include "catalog/pg_control.h"
#include "common/pg_lzcompress.h"
#include "executor/instrument.h"
#include "miscadmin.h"
#include "pg_trace.h"
#include "replication/origin.h"
#include "storage/bufmgr.h"
#include "storage/proc.h"
#include "utils/memutils.h"

/*
 * 估算存储压缩版备份块图像所需的最大缓冲区大小。
 */
#ifdef USE_LZ4
#define	LZ4_MAX_BLCKSZ		LZ4_COMPRESSBOUND(BLCKSZ)
#else
#define LZ4_MAX_BLCKSZ		0
#endif

#ifdef USE_ZSTD
#define ZSTD_MAX_BLCKSZ		ZSTD_COMPRESSBOUND(BLCKSZ)
#else
#define ZSTD_MAX_BLCKSZ		0
#endif

#define PGLZ_MAX_BLCKSZ		PGLZ_MAX_OUTPUT(BLCKSZ)

/* 存储压缩版备份块图像所需的缓冲区大小 */
#define COMPRESS_BUFSIZE	Max(Max(PGLZ_MAX_BLCKSZ, LZ4_MAX_BLCKSZ), ZSTD_MAX_BLCKSZ)

/*
 * 对于每个使用 XLogRegisterBuffer 注册的块引用，我们填充一个
 * registered_buffer 结构体。
 */
typedef struct
{
	bool		in_use;			/* 这个槽位在使用吗？ */
	uint8		flags;			/* REGBUF_* 标志 */
	RelFileNode rnode;			/* 标识关系和块 */
	ForkNumber	forkno;
	BlockNumber block;
	Page		page;			/* 页内容 */
	uint32		rdata_len;		/* rdata 链中数据的总长度 */
	XLogRecData *rdata_head;	/* 与此块注册的数据链的头部 */
	XLogRecData *rdata_tail;	/* 链中的最后一个条目，或 &rdata_head 如果为空 */

	XLogRecData bkp_rdatas[2];	/* 用于在 XLogRecordAssemble() 中保存对备份块数据的引用的临时 rdatas */

	/* 存储压缩版备份块图像的缓冲区 */
	char		compressed_page[COMPRESS_BUFSIZE];
} registered_buffer;

static registered_buffer *registered_buffers;
static int	max_registered_buffers; /* 分配的大小 */
static int	max_registered_block_id = 0;	/* 目前注册的最高 block_id + 1 */

/*
 * 一系列 XLogRecDatas 用于保存 WAL 记录的“主数据”，
 * 与 XLogRegisterData(...) 注册。
 */
static XLogRecData *mainrdata_head;
static XLogRecData *mainrdata_last = (XLogRecData *) &mainrdata_head;
static uint32 mainrdata_len;	/* 链中的总字节数 */

/* 正在进行的插入的标志 */
static uint8 curinsert_flags = 0;

/*
 * 这些用于在构造记录时保存记录头。
 * 'hdr_scratch' 不是一个普通变量，而是在初始化时使用 palloc 分配的，
 * 因为我们希望它是 MAXALIGN，并且填充字节为零。
 *
 * 为简单起见，它分配得足够大，可以容纳任何 WAL 记录的头部。
 */
static XLogRecData hdr_rdt;
static char *hdr_scratch = NULL;

#define SizeOfXlogOrigin	(sizeof(RepOriginId) + sizeof(char))
#define SizeOfXLogTransactionId	(sizeof(TransactionId) + sizeof(char))

#define HEADER_SCRATCH_SIZE \
	(SizeOfXLogRecord + \
	 MaxSizeOfXLogRecordBlockHeader * (XLR_MAX_BLOCK_ID + 1) + \
	 SizeOfXLogRecordDataHeaderLong + SizeOfXlogOrigin + \
	 SizeOfXLogTransactionId)

/*
 * 一个 XLogRecData 结构体的数组，用于保存已注册的数据。
 */
static XLogRecData *rdatas;
static int	num_rdatas;			/* 当前使用的条目 */
static int	max_rdatas;			/* 分配的大小 */

static bool begininsert_called = false;

/* 内存上下文用于保存注册的缓冲区和数据引用。 */
static MemoryContext xloginsert_cxt;

static XLogRecData *fc_XLogRecordAssemble(RmgrId fc_rmid, uint8 fc_info,
									   XLogRecPtr fc_RedoRecPtr, bool fc_doPageWrites,
									   XLogRecPtr *fc_fpw_lsn, int *fc_num_fpi,
									   bool *fc_topxid_included);
static bool fc_XLogCompressBackupBlock(char *fc_page, uint16 fc_hole_offset,
									uint16 fc_hole_length, char *fc_dest, uint16 *fc_dlen);

/*
 * 开始构造 WAL 记录。必须在调用 XLogRegister* 函数和 XLogInsert()
 * 之前调用。
 */
void XLogBeginInsert(void)
{
	Assert(max_registered_block_id == 0);
	Assert(mainrdata_last == (XLogRecData *) &mainrdata_head);
	Assert(mainrdata_len == 0);

	/* 检查我们是否应该在这里 */
	if (!XLogInsertAllowed())
		elog(ERROR, "cannot make new WAL entries during recovery");

	if (begininsert_called)
		elog(ERROR, "XLogBeginInsert was already called");

	begininsert_called = true;
}

/*
 * 确保工作区中有足够的缓冲区和数据槽位，以供后续的
 * XLogRegisterBuffer、XLogRegisterData 和 XLogRegisterBufData 调用。
 *
 * 对于大多数记录类型，总是有空间来容纳少量的缓冲区和数据块。
 * 此函数用于需要更多的特殊情况。
 */
void XLogEnsureRecordSpace(int fc_max_block_id, int fc_ndatas)
{
	int			fc_nbuffers;

	/*
	 * 必须在进入临界区之前调用，因为在临界区内分配内存可能会失败。
	 * repalloc() 会检查相同的情况，但最好在这里也检查这样即使数组
	 * 恰好已经足够大，我们也能一致地失败。
	 */
	Assert(CritSectionCount == 0);

	/* 最小值不能减少 */
	if (fc_max_block_id < XLR_NORMAL_MAX_BLOCK_ID)
		fc_max_block_id = XLR_NORMAL_MAX_BLOCK_ID;
	if (fc_ndatas < XLR_NORMAL_RDATAS)
		fc_ndatas = XLR_NORMAL_RDATAS;

	if (fc_max_block_id > XLR_MAX_BLOCK_ID)
		elog(ERROR, "maximum number of WAL record block references exceeded");
	fc_nbuffers = fc_max_block_id + 1;

	if (fc_nbuffers > max_registered_buffers)
	{
		registered_buffers = (registered_buffer *)
			repalloc(registered_buffers, sizeof(registered_buffer) * fc_nbuffers);

		/*
		 * 至少结构体中的填充字节必须被清零，因为它们包含在 WAL 数据中，
		 * 但为了整洁，将其全部初始化。
		 */
		MemSet(&registered_buffers[max_registered_buffers], 0,
			   (fc_nbuffers - max_registered_buffers) * sizeof(registered_buffer));
		max_registered_buffers = fc_nbuffers;
	}

	if (fc_ndatas > max_rdatas)
	{
		rdatas = (XLogRecData *) repalloc(rdatas, sizeof(XLogRecData) * fc_ndatas);
		max_rdatas = fc_ndatas;
	}
}

/*
 * 重置 WAL 记录构造缓冲区。
 */
void XLogResetInsertion(void)
{
	int			fc_i;

	for (fc_i = 0; fc_i < max_registered_block_id; fc_i++)
		registered_buffers[fc_i].in_use = false;

	num_rdatas = 0;
	max_registered_block_id = 0;
	mainrdata_len = 0;
	mainrdata_last = (XLogRecData *) &mainrdata_head;
	curinsert_flags = 0;
	begininsert_called = false;
}

/*
 * 在正在构造的 WAL 记录中注册对缓冲区的引用。
 * 必须为 WAL 记录的每个页面调用此函数，该页面修改了 WAL 记录。
 */
void XLogRegisterBuffer(uint8 fc_block_id, Buffer fc_buffer, uint8 fc_flags)
{
	registered_buffer *fc_regbuf;

	/* NO_IMAGE 在 FORCE_IMAGE 下没有意义 */
	Assert(!((fc_flags & REGBUF_FORCE_IMAGE) && (fc_flags & (REGBUF_NO_IMAGE))));
	Assert(begininsert_called);

	if (fc_block_id >= max_registered_block_id)
	{
		if (fc_block_id >= max_registered_buffers)
			elog(ERROR, "too many registered buffers");
		max_registered_block_id = fc_block_id + 1;
	}

	fc_regbuf = &registered_buffers[fc_block_id];

	BufferGetTag(fc_buffer, &fc_regbuf->rnode, &fc_regbuf->forkno, &fc_regbuf->block);
	fc_regbuf->page = BufferGetPage(fc_buffer);
	fc_regbuf->flags = fc_flags;
	fc_regbuf->rdata_tail = (XLogRecData *) &fc_regbuf->rdata_head;
	fc_regbuf->rdata_len = 0;

	/*
	 * 检查该页面是否已经与某个其他
	 * block_id 注册。
	 */
#ifdef USE_ASSERT_CHECKING
	{
		int			fc_i;

		for (fc_i = 0; fc_i < max_registered_block_id; fc_i++)
		{
			registered_buffer *fc_regbuf_old = &registered_buffers[fc_i];

			if (fc_i == fc_block_id || !fc_regbuf_old->in_use)
				continue;

			Assert(!RelFileNodeEquals(fc_regbuf_old->rnode, fc_regbuf->rnode) ||
				   fc_regbuf_old->forkno != fc_regbuf->forkno ||
				   fc_regbuf_old->block != fc_regbuf->block);
		}
	}
#endif

	fc_regbuf->in_use = true;
}

/*
 * 类似于 XLogRegisterBuffer，但用于注册一个不在
 * 共享缓冲池中的块（即当你没有 Buffer 时）。
 */
void XLogRegisterBlock(uint8 fc_block_id, RelFileNode *fc_rnode, ForkNumber fc_forknum,
				  BlockNumber fc_blknum, Page fc_page, uint8 fc_flags)
{
	registered_buffer *fc_regbuf;

	Assert(begininsert_called);

	if (fc_block_id >= max_registered_block_id)
		max_registered_block_id = fc_block_id + 1;

	if (fc_block_id >= max_registered_buffers)
		elog(ERROR, "too many registered buffers");

	fc_regbuf = &registered_buffers[fc_block_id];

	fc_regbuf->rnode = *fc_rnode;
	fc_regbuf->forkno = fc_forknum;
	fc_regbuf->block = fc_blknum;
	fc_regbuf->page = fc_page;
	fc_regbuf->flags = fc_flags;
	fc_regbuf->rdata_tail = (XLogRecData *) &fc_regbuf->rdata_head;
	fc_regbuf->rdata_len = 0;

	/*
	 * 检查该页面是否已经与某个其他
	 * block_id 注册。
	 */
#ifdef USE_ASSERT_CHECKING
	{
		int			fc_i;

		for (fc_i = 0; fc_i < max_registered_block_id; fc_i++)
		{
			registered_buffer *fc_regbuf_old = &registered_buffers[fc_i];

			if (fc_i == fc_block_id || !fc_regbuf_old->in_use)
				continue;

			Assert(!RelFileNodeEquals(fc_regbuf_old->rnode, fc_regbuf->rnode) ||
				   fc_regbuf_old->forkno != fc_regbuf->forkno ||
				   fc_regbuf_old->block != fc_regbuf->block);
		}
	}
#endif

	fc_regbuf->in_use = true;
}

/*
 * 向正在构建的 WAL 记录添加数据。
 *
 * 数据将附加到“主块”，可通过
 * XLogRecGetData() 回复。
 */
void XLogRegisterData(char *fc_data, int fc_len)
{
	XLogRecData *fc_rdata;

	Assert(begininsert_called);

	if (num_rdatas >= max_rdatas)
		elog(ERROR, "too much WAL data");
	fc_rdata = &rdatas[num_rdatas++];

	fc_rdata->data = fc_data;
	fc_rdata->len = fc_len;

	/*
	 * 我们使用 mainrdata_last 指针来跟踪链的末尾，因此这里不
	 * 需要清除 'next'。
	 */

	mainrdata_last->next = fc_rdata;
	mainrdata_last = fc_rdata;

	mainrdata_len += fc_len;
}

/*
 * 向正在构建的 WAL 记录添加缓冲区特定的数据。
 *
 * Block_id 必须引用先前使用
 * XLogRegisterBuffer() 注册的块。如果对同一
 * block_id 多次调用此函数，数据将附加。
 *
 * 每个块可以注册的最大数据量为 65535
 * 字节。这个数量应该足够；如果你需要超过 BLCKSZ 字节来
 * 重构页面的更改，不如直接记录其完整副本。（与块无关的“主数据”没有
 * 限制）
 */
void XLogRegisterBufData(uint8 fc_block_id, char *fc_data, int fc_len)
{
	registered_buffer *fc_regbuf;
	XLogRecData *fc_rdata;

	Assert(begininsert_called);

	/* 查找已注册的缓冲区结构 */
	fc_regbuf = &registered_buffers[fc_block_id];
	if (!fc_regbuf->in_use)
		elog(ERROR, "no block with id %d registered with WAL insertion",
			 fc_block_id);

	if (num_rdatas >= max_rdatas)
		elog(ERROR, "too much WAL data");
	fc_rdata = &rdatas[num_rdatas++];

	fc_rdata->data = fc_data;
	fc_rdata->len = fc_len;

	fc_regbuf->rdata_tail->next = fc_rdata;
	fc_regbuf->rdata_tail = fc_rdata;
	fc_regbuf->rdata_len += fc_len;
}

/*
 * 为即将到来的 WAL 记录设置插入状态标志。
 *
 * 这里可以使用的标志有：
 * - XLOG_INCLUDE_ORIGIN，决定是否应在记录中包含
 *	复制源。
 * - XLOG_MARK_UNIMPORTANT，表示记录 对于
 *	持久性不重要，这样可以避免触发 WAL 归档和其他
 *	后台活动。
 */
void XLogSetRecordFlags(uint8 fc_flags)
{
	Assert(begininsert_called);
	curinsert_flags |= fc_flags;
}

/*
 * 插入一个具有指定 RMID 和信息字节的 XLOG 记录，其
 * 记录体为早先通过 XLogRegister* 调用注册的数据和缓冲区引用。
 *
 * 返回指向记录末尾的 XLOG 指针（下一个记录的开始）。
 * 这可以用作被记录操作影响的数据页面的 LSN。
 * （LSN 是 XLOG 需要刷新到磁盘的点
 * ，在数据页面可以被写出之前。 这实现了基本的
 * WAL 规则“在数据之前写日志”。）
 */
XLogRecPtr XLogInsert(RmgrId fc_rmid, uint8 fc_info)
{
	XLogRecPtr	fc_EndPos;

	/* 必须调用 XLogBeginInsert()。 */
	if (!begininsert_called)
		elog(ERROR, "XLogBeginInsert was not called");

	/*
	 * 调用者可以设置 rmgr 位、XLR_SPECIAL_REL_UPDATE 和
	 * XLR_CHECK_CONSISTENCY；其余的保留供我使用。
	 */
	if ((fc_info & ~(XLR_RMGR_INFO_MASK |
				  XLR_SPECIAL_REL_UPDATE |
				  XLR_CHECK_CONSISTENCY)) != 0)
		elog(PANIC, "invalid xlog info mask %02X", fc_info);

	TRACE_POSTGRESQL_WAL_INSERT(rmid, info);

	/*
	 * 在引导模式下，我们实际上不记录任何内容，但 XLOG 资源；
	 * 返回一个虚假的记录指针。
	 */
	if (IsBootstrapProcessingMode() && fc_rmid != RM_XLOG_ID)
	{
		XLogResetInsertion();
		fc_EndPos = SizeOfXLogLongPHD; /* 第一个检查点记录的开始 */
		return fc_EndPos;
	}

	do
	{
		XLogRecPtr	fc_RedoRecPtr;
		bool		fc_doPageWrites;
		bool		fc_topxid_included = false;
		XLogRecPtr	fc_fpw_lsn;
		XLogRecData *fc_rdt;
		int			fc_num_fpi = 0;

		/*
		 * 获取决定是否执行全页写入所需的值。由于
		 * 我们还没有获得插入锁，这些值可能会在我们
		 * 之下发生变化，但 XLogInsertRecord 在获得锁后将重新检查它们。
		 */
		GetFullPageWriteInfo(&fc_RedoRecPtr, &fc_doPageWrites);

		fc_rdt = fc_XLogRecordAssemble(fc_rmid, fc_info, fc_RedoRecPtr, fc_doPageWrites,
								 &fc_fpw_lsn, &fc_num_fpi, &fc_topxid_included);

		fc_EndPos = XLogInsertRecord(fc_rdt, fc_fpw_lsn, curinsert_flags, fc_num_fpi,
								  fc_topxid_included);
	} while (fc_EndPos == InvalidXLogRecPtr);

	XLogResetInsertion();

	return fc_EndPos;
}

/*
 * 将注册的数据和缓冲区组合成一个WAL记录，形成一个
 * XLogRecData链，准备插入到XLogInsertRecord()中。
 *
 * 除xl_prev字段外，记录头字段已填充。计算的CRC尚未包括记录头。
 *
 * 如果有任何注册的缓冲区，并且没有对所有缓冲区进行全页图像的采集，
 * *fpw_lsn被设置为这些页面中最低的LSN。这表示组合的记录仅适用于在
 * 假设RedoRecPtr和doPageWrites值是最新的情况下进行插入。
 *
 * 如果当前子事务记录了顶层事务ID，则设置*topxid_included。
 */
static XLogRecData *
fc_XLogRecordAssemble(RmgrId fc_rmid, uint8 fc_info,
				   XLogRecPtr fc_RedoRecPtr, bool fc_doPageWrites,
				   XLogRecPtr *fc_fpw_lsn, int *fc_num_fpi, bool *fc_topxid_included)
{
	XLogRecData *fc_rdt;
	uint32		fc_total_len = 0;
	int			fc_block_id;
	pg_crc32c	fc_rdata_crc;
	registered_buffer *fc_prev_regbuf = NULL;
	XLogRecData *fc_rdt_datas_last;
	XLogRecord *fc_rechdr;
	char	   *fc_scratch = hdr_scratch;

	/*
	 * 注意：可以对同一记录多次调用此函数。
	 * 我们对下面的rdata链所做的所有修改必须处理这一点。
	 */

	/* 记录以固定大小的头部开始 */
	fc_rechdr = (XLogRecord *) fc_scratch;
	fc_scratch += SizeOfXLogRecord;

	hdr_rdt.next = NULL;
	fc_rdt_datas_last = &hdr_rdt;
	hdr_rdt.data = hdr_scratch;

	/*
	 * 如果用户在查找，则强制执行此记录的一致性检查。
	 * 在例程开始时执行此操作，以便给XLogInsert()的调用者
	 * 传递XLR_CHECK_CONSISTENCY的可能性。
	 */
	if (wal_consistency_checking[fc_rmid])
		fc_info |= XLR_CHECK_CONSISTENCY;

	/*
	 * 创建一个包含所有块引用的所有数据部分的rdata链。
	 * 这包括全页图像的数据。同时在临时缓冲区附加
	 * 块引用的头部。
	 */
	*fc_fpw_lsn = InvalidXLogRecPtr;
	for (fc_block_id = 0; fc_block_id < max_registered_block_id; fc_block_id++)
	{
		registered_buffer *fc_regbuf = &registered_buffers[fc_block_id];
		bool		fc_needs_backup;
		bool		fc_needs_data;
		XLogRecordBlockHeader fc_bkpb;
		XLogRecordBlockImageHeader fc_bimg;
		XLogRecordBlockCompressHeader fc_cbimg = {0};
		bool		fc_samerel;
		bool		fc_is_compressed = false;
		bool		fc_include_image;

		if (!fc_regbuf->in_use)
			continue;

		/* 确定此块是否需要备份 */
		if (fc_regbuf->flags & REGBUF_FORCE_IMAGE)
			fc_needs_backup = true;
		else if (fc_regbuf->flags & REGBUF_NO_IMAGE)
			fc_needs_backup = false;
		else if (!fc_doPageWrites)
			fc_needs_backup = false;
		else
		{
			/*
			 * 我们假设页面LSN是*每*个可以传递给XLogInsert的
			 * 页面的第一条数据，无论它是否具有标准页面布局。
			 */
			XLogRecPtr	fc_page_lsn = PageGetLSN(fc_regbuf->page);

			fc_needs_backup = (fc_page_lsn <= fc_RedoRecPtr);
			if (!fc_needs_backup)
			{
				if (*fc_fpw_lsn == InvalidXLogRecPtr || fc_page_lsn < *fc_fpw_lsn)
					*fc_fpw_lsn = fc_page_lsn;
			}
		}

		/* 确定缓冲区数据是否需要包含 */
		if (fc_regbuf->rdata_len == 0)
			fc_needs_data = false;
		else if ((fc_regbuf->flags & REGBUF_KEEP_DATA) != 0)
			fc_needs_data = true;
		else
			fc_needs_data = !fc_needs_backup;

		fc_bkpb.id = fc_block_id;
		fc_bkpb.fork_flags = fc_regbuf->forkno;
		fc_bkpb.data_length = 0;

		if ((fc_regbuf->flags & REGBUF_WILL_INIT) == REGBUF_WILL_INIT)
			fc_bkpb.fork_flags |= BKPBLOCK_WILL_INIT;

		/*
		 * 如果needs_backup为真或当前资源管理器启用了WAL检查，
		 * 则记录当前块的全页写入。
		 */
		fc_include_image = fc_needs_backup || (fc_info & XLR_CHECK_CONSISTENCY) != 0;

		if (fc_include_image)
		{
			Page		fc_page = fc_regbuf->page;
			uint16		fc_compressed_len = 0;

			/*
			 * 页面需要备份，因此计算其孔长度和偏移量。
			 */
			if (fc_regbuf->flags & REGBUF_STANDARD)
			{
				/* 假设我们可以省略pd_lower和pd_upper之间的数据 */
				uint16		fc_lower = ((PageHeader) fc_page)->pd_lower;
				uint16		fc_upper = ((PageHeader) fc_page)->pd_upper;

				if (fc_lower >= SizeOfPageHeaderData &&
					fc_upper > fc_lower &&
					fc_upper <= BLCKSZ)
				{
					fc_bimg.hole_offset = fc_lower;
					fc_cbimg.hole_length = fc_upper - fc_lower;
				}
				else
				{
					/* 没有“孔”可以移除 */
					fc_bimg.hole_offset = 0;
					fc_cbimg.hole_length = 0;
				}
			}
			else
			{
				/* 不是标准页面头，不要尝试消除“孔” */
				fc_bimg.hole_offset = 0;
				fc_cbimg.hole_length = 0;
			}

			/*
			 * 如果启用了wal_compression，则尝试压缩块图像
			 */
			if (wal_compression != WAL_COMPRESSION_NONE)
			{
				fc_is_compressed =
					fc_XLogCompressBackupBlock(fc_page, fc_bimg.hole_offset,
											fc_cbimg.hole_length,
											fc_regbuf->compressed_page,
											&fc_compressed_len);
			}

			/*
			 * 填充XLogRecordBlockHeader结构中的剩余字段
			 */
			fc_bkpb.fork_flags |= BKPBLOCK_HAS_IMAGE;

			/* 报告为WAL记录构造的全页图像 */
			*fc_num_fpi += 1;

			/*
			 * 为页面内容构造XLogRecData条目。
			 */
			fc_rdt_datas_last->next = &fc_regbuf->bkp_rdatas[0];
			fc_rdt_datas_last = fc_rdt_datas_last->next;

			fc_bimg.bimg_info = (fc_cbimg.hole_length == 0) ? 0 : BKPIMAGE_HAS_HOLE;

			/*
			 * 如果当前资源管理器为此WAL记录启用了WAL一致性检查，
			 * 则将在记录中包含为已修改块的全页图像。
			 * 在重做期间，仅当设置了BKPIMAGE_APPLY时，才会重放全页。
			 */
			if (fc_needs_backup)
				fc_bimg.bimg_info |= BKPIMAGE_APPLY;

			if (fc_is_compressed)
			{
				/* 当前的压缩方法存储在WAL记录中 */
				fc_bimg.length = fc_compressed_len;

				/* 设置此块的压缩方法 */
				switch ((WalCompression) wal_compression)
				{
					case WAL_COMPRESSION_PGLZ:
						fc_bimg.bimg_info |= BKPIMAGE_COMPRESS_PGLZ;
						break;

					case WAL_COMPRESSION_LZ4:
#ifdef USE_LZ4
						fc_bimg.bimg_info |= BKPIMAGE_COMPRESS_LZ4;
#else
						elog(ERROR, "LZ4 is not supported by this build");
#endif
						break;

					case WAL_COMPRESSION_ZSTD:
#ifdef USE_ZSTD
						fc_bimg.bimg_info |= BKPIMAGE_COMPRESS_ZSTD;
#else
						elog(ERROR, "zstd is not supported by this build");
#endif
						break;

					case WAL_COMPRESSION_NONE:
						Assert(false);	/* 不会发生 */
						break;
						/* 没有默认情况，以便编译器会发出警告 */
				}

				fc_rdt_datas_last->data = fc_regbuf->compressed_page;
				fc_rdt_datas_last->len = fc_compressed_len;
			}
			else
			{
				fc_bimg.length = BLCKSZ - fc_cbimg.hole_length;

				if (fc_cbimg.hole_length == 0)
				{
					fc_rdt_datas_last->data = fc_page;
					fc_rdt_datas_last->len = BLCKSZ;
				}
				else
				{
					/* 必须跳过孔 */
					fc_rdt_datas_last->data = fc_page;
					fc_rdt_datas_last->len = fc_bimg.hole_offset;

					fc_rdt_datas_last->next = &fc_regbuf->bkp_rdatas[1];
					fc_rdt_datas_last = fc_rdt_datas_last->next;

					fc_rdt_datas_last->data =
						fc_page + (fc_bimg.hole_offset + fc_cbimg.hole_length);
					fc_rdt_datas_last->len =
						BLCKSZ - (fc_bimg.hole_offset + fc_cbimg.hole_length);
				}
			}

			fc_total_len += fc_bimg.length;
		}

		if (fc_needs_data)
		{
			/*
			 * 将调用方提供的此缓冲区的rdata链链接到
			 * 整体列表中。
			 */
			fc_bkpb.fork_flags |= BKPBLOCK_HAS_DATA;
			fc_bkpb.data_length = fc_regbuf->rdata_len;
			fc_total_len += fc_regbuf->rdata_len;

			fc_rdt_datas_last->next = fc_regbuf->rdata_head;
			fc_rdt_datas_last = fc_regbuf->rdata_tail;
		}

		if (fc_prev_regbuf && RelFileNodeEquals(fc_regbuf->rnode, fc_prev_regbuf->rnode))
		{
			fc_samerel = true;
			fc_bkpb.fork_flags |= BKPBLOCK_SAME_REL;
		}
		else
			fc_samerel = false;
		fc_prev_regbuf = fc_regbuf;

		/* 好的，将标题复制到临时缓冲区 */
		memcpy(fc_scratch, &fc_bkpb, SizeOfXLogRecordBlockHeader);
		fc_scratch += SizeOfXLogRecordBlockHeader;
		if (fc_include_image)
		{
			memcpy(fc_scratch, &fc_bimg, SizeOfXLogRecordBlockImageHeader);
			fc_scratch += SizeOfXLogRecordBlockImageHeader;
			if (fc_cbimg.hole_length != 0 && fc_is_compressed)
			{
				memcpy(fc_scratch, &fc_cbimg,
					   SizeOfXLogRecordBlockCompressHeader);
				fc_scratch += SizeOfXLogRecordBlockCompressHeader;
			}
		}
		if (!fc_samerel)
		{
			memcpy(fc_scratch, &fc_regbuf->rnode, sizeof(RelFileNode));
			fc_scratch += sizeof(RelFileNode);
		}
		memcpy(fc_scratch, &fc_regbuf->block, sizeof(BlockNumber));
		fc_scratch += sizeof(BlockNumber);
	}

	/* 随后是记录的来源（如果有的话） */
	if ((curinsert_flags & XLOG_INCLUDE_ORIGIN) &&
		replorigin_session_origin != InvalidRepOriginId)
	{
		*(fc_scratch++) = (char) XLR_BLOCK_ID_ORIGIN;
		memcpy(fc_scratch, &replorigin_session_origin, sizeof(replorigin_session_origin));
		fc_scratch += sizeof(replorigin_session_origin);
	}

	/* 随后是顶层 XID，如果之前的记录中没有包含的话 */
	if (IsSubxactTopXidLogPending())
	{
		TransactionId fc_xid = GetTopTransactionIdIfAny();

		/* 设置标志，表示顶层 xid 已包含在 WAL 中 */
		*fc_topxid_included = true;

		*(fc_scratch++) = (char) XLR_BLOCK_ID_TOPLEVEL_XID;
		memcpy(fc_scratch, &fc_xid, sizeof(TransactionId));
		fc_scratch += sizeof(TransactionId);
	}

	/* 随后是主数据（如果有的话） */
	if (mainrdata_len > 0)
	{
		if (mainrdata_len > 255)
		{
			*(fc_scratch++) = (char) XLR_BLOCK_ID_DATA_LONG;
			memcpy(fc_scratch, &mainrdata_len, sizeof(uint32));
			fc_scratch += sizeof(uint32);
		}
		else
		{
			*(fc_scratch++) = (char) XLR_BLOCK_ID_DATA_SHORT;
			*(fc_scratch++) = (uint8) mainrdata_len;
		}
		fc_rdt_datas_last->next = mainrdata_head;
		fc_rdt_datas_last = mainrdata_last;
		fc_total_len += mainrdata_len;
	}
	fc_rdt_datas_last->next = NULL;

	hdr_rdt.len = (fc_scratch - hdr_scratch);
	fc_total_len += hdr_rdt.len;

	/*
	 * 计算数据的 CRC
	 *
	 * 注意，记录头最初没有添加到 CRC 中，因为我们
	 * 还不知道 prev-link。因此，CRC 将表示
	 * 整个记录的 CRC，顺序为：rdata，然后是备份块，然后是记录头。
	 */
	INIT_CRC32C(fc_rdata_crc);
	COMP_CRC32C(fc_rdata_crc, hdr_scratch + SizeOfXLogRecord, hdr_rdt.len - SizeOfXLogRecord);
	for (fc_rdt = hdr_rdt.next; fc_rdt != NULL; fc_rdt = fc_rdt->next)
		COMP_CRC32C(fc_rdata_crc, fc_rdt->data, fc_rdt->len);

	/*
	 * 填充记录头中的字段。Prev-link 将在我们知道
	 * 记录将在 WAL 中插入的位置后再填写。CRC
	 * 尚未包含记录头。
	 */
	fc_rechdr->xl_xid = GetCurrentTransactionIdIfAny();
	fc_rechdr->xl_tot_len = fc_total_len;
	fc_rechdr->xl_info = fc_info;
	fc_rechdr->xl_rmid = fc_rmid;
	fc_rechdr->xl_prev = InvalidXLogRecPtr;
	fc_rechdr->xl_crc = fc_rdata_crc;

	return &hdr_rdt;
}

/*
 * 创建备份块图像的压缩版本。
 *
 * 如果压缩失败（即，压缩结果实际上
 * 大于原始数据），则返回 false。否则，返回 true 并设置 'dlen' 为
 * 压缩块图像的长度。
 */
static bool fc_XLogCompressBackupBlock(char *fc_page, uint16 fc_hole_offset, uint16 fc_hole_length,
						char *fc_dest, uint16 *fc_dlen)
{
	int32		fc_orig_len = BLCKSZ - fc_hole_length;
	int32		fc_len = -1;
	int32		fc_extra_bytes = 0;
	char	   *fc_source;
	PGAlignedBlock fc_tmp;

	if (fc_hole_length != 0)
	{
		/* 必须跳过孔 */
		fc_source = fc_tmp.data;
		memcpy(fc_source, fc_page, fc_hole_offset);
		memcpy(fc_source + fc_hole_offset,
			   fc_page + (fc_hole_offset + fc_hole_length),
			   BLCKSZ - (fc_hole_length + fc_hole_offset));

		/*
		 * 如果空洞存在，则需要在 WAL 记录中存储额外数据
		 * 用于块图像的压缩版本。
		 */
		fc_extra_bytes = SizeOfXLogRecordBlockCompressHeader;
	}
	else
		fc_source = fc_page;

	switch ((WalCompression) wal_compression)
	{
		case WAL_COMPRESSION_PGLZ:
			fc_len = pglz_compress(fc_source, fc_orig_len, fc_dest, PGLZ_strategy_default);
			break;

		case WAL_COMPRESSION_LZ4:
#ifdef USE_LZ4
			fc_len = LZ4_compress_default(fc_source, fc_dest, fc_orig_len,
									   COMPRESS_BUFSIZE);
			if (fc_len <= 0)
				fc_len = -1;		/* 失败 */
#else
			elog(ERROR, "LZ4 is not supported by this build");
#endif
			break;

		case WAL_COMPRESSION_ZSTD:
#ifdef USE_ZSTD
			fc_len = ZSTD_compress(fc_dest, COMPRESS_BUFSIZE, fc_source, fc_orig_len,
								ZSTD_CLEVEL_DEFAULT);
			if (ZSTD_isError(fc_len))
				fc_len = -1;		/* 失败 */
#else
			elog(ERROR, "zstd is not supported by this build");
#endif
			break;

		case WAL_COMPRESSION_NONE:
			Assert(false);		/* 不会发生 */
			break;
			/* 没有默认情况，以便编译器会发出警告 */
	}

	/*
	 * 即使压缩报告成功，我们也要重新检查实际大小，
	 * 并查看压缩节省的字节数是否大于压缩版本所需的额外数据长度。
	 */
	if (fc_len >= 0 &&
		fc_len + fc_extra_bytes < fc_orig_len)
	{
		*fc_dlen = (uint16) fc_len;	/* 成功的压缩 */
		return true;
	}
	return false;
}

/*
 * 确定所引用的缓冲区是否需要备份。
 *
 * 由于我们还没有插入锁，fullPageWrites 和 forcePageWrites
 * 可能会在稍后更改，因此结果仅应用于优化目的。
 */
bool XLogCheckBufferNeedsBackup(Buffer fc_buffer)
{
	XLogRecPtr	fc_RedoRecPtr;
	bool		fc_doPageWrites;
	Page		fc_page;

	GetFullPageWriteInfo(&fc_RedoRecPtr, &fc_doPageWrites);

	fc_page = BufferGetPage(fc_buffer);

	if (fc_doPageWrites && PageGetLSN(fc_page) <= fc_RedoRecPtr)
		return true;			/* 缓冲区需要备份 */

	return false;				/* 缓冲区不需要备份 */
}

/*
 * 在设置提示时，如有需要请写入备份块。注意，
 * 这可能会针对多种页面类型调用，而不仅仅是堆。
 *
 * 在仅持有缓冲区内容的共享锁时可调用。
 *
 * 我们无法使用普通的备份块机制，因为那依赖于
 * 缓冲区的独占锁定。由于一些修改（设置 LSN、提示
 * 位）在共享锁缓冲区中是允许的，这可能导致 WAL 校验和
 * 失败。因此，我们改为复制页面并将复制的数据插入为正常
 * 的记录数据。
 *
 * 只有在页面还未在此检查点轮次中完全写入时，我们才
 * 需要做某些操作。如果我们必须写入，则返回插入的 WAL
 * 记录的 LSN，否则返回 InvalidXLogRecPtr。
 *
 * 可能存在多个并发后端尝试写入 WAL
 * 记录的情况。在这种情况下，多个相同块的副本将
 * 被不同的后端记录在独立的 WAL 记录中，不过从
 * 正确性角度来看，这仍然是可以的。
 */
XLogRecPtr XLogSaveBufferForHint(Buffer fc_buffer, bool fc_buffer_std)
{
	XLogRecPtr	fc_recptr = InvalidXLogRecPtr;
	XLogRecPtr	fc_lsn;
	XLogRecPtr	fc_RedoRecPtr;

	/*
	 * 确保没有检查点可以改变我们对 RedoRecPtr 的视图。
	 */
	Assert((MyProc->delayChkptFlags & DELAY_CHKPT_START) != 0);

	/*
	 * 更新 RedoRecPtr，以便我们可以做出正确的决策
	 */
	fc_RedoRecPtr = GetRedoRecPtr();

	/*
	 * 我们假设页面 LSN 是可以传递给 XLogInsert 的 *每个* 页面上的第一条数据，无论它是否具有标准页面布局。由于我们仅在页面上持有共享锁，因此在查看 LSN 时必须获取缓冲区头部锁。
	 */
	fc_lsn = BufferGetLSNAtomic(fc_buffer);

	if (fc_lsn <= fc_RedoRecPtr)
	{
		int			fc_flags = 0;
		PGAlignedBlock fc_copied_buffer;
		char	   *fc_origdata = (char *) BufferGetBlock(fc_buffer);
		RelFileNode fc_rnode;
		ForkNumber	fc_forkno;
		BlockNumber fc_blkno;

		/*
		 * 复制缓冲区，以便我们不必担心并发的提示位或 LSN 更新。我们假设 pd_lower/upper 不能在没有排他锁的情况下更改，因此内容 bkp 不是竞争的。
		 */
		if (fc_buffer_std)
		{
			/* 假设我们可以省略pd_lower和pd_upper之间的数据 */
			Page		fc_page = BufferGetPage(fc_buffer);
			uint16		fc_lower = ((PageHeader) fc_page)->pd_lower;
			uint16		fc_upper = ((PageHeader) fc_page)->pd_upper;

			memcpy(fc_copied_buffer.data, fc_origdata, fc_lower);
			memcpy(fc_copied_buffer.data + fc_upper, fc_origdata + fc_upper, BLCKSZ - fc_upper);
		}
		else
			memcpy(fc_copied_buffer.data, fc_origdata, BLCKSZ);

		XLogBeginInsert();

		if (fc_buffer_std)
			fc_flags |= REGBUF_STANDARD;

		BufferGetTag(fc_buffer, &fc_rnode, &fc_forkno, &fc_blkno);
		XLogRegisterBlock(0, &fc_rnode, fc_forkno, fc_blkno, fc_copied_buffer.data, fc_flags);

		fc_recptr = XLogInsert(RM_XLOG_ID, XLOG_FPI_FOR_HINT);
	}

	return fc_recptr;
}

/*
 * 写入包含页面完整图像的 WAL 记录。调用者负责在调用此例程后将页面写入磁盘。
 *
 * 注意：如果您使用此函数，则应该在私有内存中构建页面并直接写入 smgr。如果您使用缓冲区，请改为调用 log_newpage_buffer。
 *
 * 如果页面遵循标准页面布局，具有 PageHeader 和 pd_lower 与 pd_upper 之间的未使用空间，则将 'page_std' 设置为 true。这允许将未使用的空间从 WAL 记录中省略，从而使其更小。
 */
XLogRecPtr log_newpage(RelFileNode *fc_rnode, ForkNumber fc_forkNum, BlockNumber fc_blkno,
			Page fc_page, bool fc_page_std)
{
	int			fc_flags;
	XLogRecPtr	fc_recptr;

	fc_flags = REGBUF_FORCE_IMAGE;
	if (fc_page_std)
		fc_flags |= REGBUF_STANDARD;

	XLogBeginInsert();
	XLogRegisterBlock(0, fc_rnode, fc_forkNum, fc_blkno, fc_page, fc_flags);
	fc_recptr = XLogInsert(RM_XLOG_ID, XLOG_FPI);

	/*
	 * 页面可能未初始化。如果是这样，我们无法设置 LSN，因为这会破坏页面。
	 */
	if (!PageIsNew(fc_page))
	{
		PageSetLSN(fc_page, fc_recptr);
	}

	return fc_recptr;
}

/*
 * 类似于 log_newpage()，但允许在一次操作中记录多个页面。它比分别为每个页面调用 log_newpage() 更有效，因为我们可以在单个 WAL 记录中写入多个页面。
 */
void log_newpages(RelFileNode *fc_rnode, ForkNumber fc_forkNum, int fc_num_pages,
			 BlockNumber *fc_blknos, Page *fc_pages, bool fc_page_std)
{
	int			fc_flags;
	XLogRecPtr	fc_recptr;
	int			fc_i;
	int			fc_j;

	fc_flags = REGBUF_FORCE_IMAGE;
	if (fc_page_std)
		fc_flags |= REGBUF_STANDARD;

	/*
	 * 遍历所有页面。它们被收集到 XLR_MAX_BLOCK_ID 页面批次中，每个批次写入一个 WAL 记录。
	 */
	XLogEnsureRecordSpace(XLR_MAX_BLOCK_ID - 1, 0);

	fc_i = 0;
	while (fc_i < fc_num_pages)
	{
		int			fc_batch_start = fc_i;
		int			fc_nbatch;

		XLogBeginInsert();

		fc_nbatch = 0;
		while (fc_nbatch < XLR_MAX_BLOCK_ID && fc_i < fc_num_pages)
		{
			XLogRegisterBlock(fc_nbatch, fc_rnode, fc_forkNum, fc_blknos[fc_i], fc_pages[fc_i], fc_flags);
			fc_i++;
			fc_nbatch++;
		}

		fc_recptr = XLogInsert(RM_XLOG_ID, XLOG_FPI);

		for (fc_j = fc_batch_start; fc_j < fc_i; fc_j++)
		{
			/*
			 * 页面可能未初始化。如果是这样，我们无法设置 LSN，因为这会破坏页面。
			 */
			if (!PageIsNew(fc_pages[fc_j]))
			{
				PageSetLSN(fc_pages[fc_j], fc_recptr);
			}
		}
	}
}

/*
 * 写入包含页面完整图像的 WAL 记录。
 *
 * 调用者在调用此函数之前应初始化缓冲区并标记为脏。此函数将设置页面 LSN。
 *
 * 如果页面遵循标准页面布局，具有 PageHeader 和 pd_lower 与 pd_upper 之间的未使用空间，则将 'page_std' 设置为 true。这允许将未使用的空间从 WAL 记录中省略，从而使其更小。
 */
XLogRecPtr log_newpage_buffer(Buffer fc_buffer, bool fc_page_std)
{
	Page		fc_page = BufferGetPage(fc_buffer);
	RelFileNode fc_rnode;
	ForkNumber	fc_forkNum;
	BlockNumber fc_blkno;

	/* 共享缓冲区应在关键区域内进行修改。 */
	Assert(CritSectionCount > 0);

	BufferGetTag(fc_buffer, &fc_rnode, &fc_forkNum, &fc_blkno);

	return log_newpage(&fc_rnode, fc_forkNum, fc_blkno, fc_page, fc_page_std);
}

/*
 * 在一个关系中记录一系列块到WAL日志。
 *
 * 将所有页面的图像，块编号为'startblk' <= X < 'endblk'，写入
 * WAL。如果范围较大，将在多个WAL记录中执行此操作。
 *
 * 如果所有页面遵循标准页面布局，具有PageHeader和pd_lower与pd_upper之间的未使用
 * 空间，则将'page_std'设置为true。这允许在WAL记录中省略未使用的空间，从而使它们更小。
 *
 * 注意：此函数对页面获取独占锁。通常，这用于新建的关系，
 * 调用者持有其上的AccessExclusiveLock，因此没有其他后台可以同时访问它。
 * 如果情况不是这样，您必须确保这不会通过其他方式造成死锁。
 */
void log_newpage_range(Relation fc_rel, ForkNumber fc_forkNum,
				  BlockNumber fc_startblk, BlockNumber fc_endblk,
				  bool fc_page_std)
{
	int			fc_flags;
	BlockNumber fc_blkno;

	fc_flags = REGBUF_FORCE_IMAGE;
	if (fc_page_std)
		fc_flags |= REGBUF_STANDARD;

	/*
	 * 遍历范围内的所有页面。它们被收集为
	 * XLR_MAX_BLOCK_ID页面的批次，并为每个批次写入
	 * 一个WAL记录。
	 */
	XLogEnsureRecordSpace(XLR_MAX_BLOCK_ID - 1, 0);

	fc_blkno = fc_startblk;
	while (fc_blkno < fc_endblk)
	{
		Buffer		fc_bufpack[XLR_MAX_BLOCK_ID];
		XLogRecPtr	fc_recptr;
		int			fc_nbufs;
		int			fc_i;

		CHECK_FOR_INTERRUPTS();

		/* 收集一批块。 */
		fc_nbufs = 0;
		while (fc_nbufs < XLR_MAX_BLOCK_ID && fc_blkno < fc_endblk)
		{
			Buffer		fc_buf = ReadBufferExtended(fc_rel, fc_forkNum, fc_blkno,
												 RBM_NORMAL, NULL);

			LockBuffer(fc_buf, BUFFER_LOCK_EXCLUSIVE);

			/*
			 * 完全空的页面不会进行WAL记录。
			 * 写入WAL记录将改变LSN，我们不希望那样。
			 * 我们希望页面保持为空。
			 */
			if (!PageIsNew(BufferGetPage(fc_buf)))
				fc_bufpack[fc_nbufs++] = fc_buf;
			else
				UnlockReleaseBuffer(fc_buf);
			fc_blkno++;
		}

		/* 如果所有剩余块都是空的，则没有更多操作。 */
		if (fc_nbufs == 0)
			break;

		/* 为此批次写入WAL记录。 */
		XLogBeginInsert();

		START_CRIT_SECTION();
		for (fc_i = 0; fc_i < fc_nbufs; fc_i++)
		{
			XLogRegisterBuffer(fc_i, fc_bufpack[fc_i], fc_flags);
			MarkBufferDirty(fc_bufpack[fc_i]);
		}

		fc_recptr = XLogInsert(RM_XLOG_ID, XLOG_FPI);

		for (fc_i = 0; fc_i < fc_nbufs; fc_i++)
		{
			PageSetLSN(BufferGetPage(fc_bufpack[fc_i]), fc_recptr);
			UnlockReleaseBuffer(fc_bufpack[fc_i]);
		}
		END_CRIT_SECTION();
	}
}

/*
 * 为WAL记录构建分配所需的工作缓冲区。
 */
void InitXLogInsert(void)
{
	/* 初始化工作区域 */
	if (xloginsert_cxt == NULL)
	{
		xloginsert_cxt = AllocSetContextCreate(TopMemoryContext,
											   "WAL record construction",
											   ALLOCSET_DEFAULT_SIZES);
	}

	if (registered_buffers == NULL)
	{
		registered_buffers = (registered_buffer *)
			MemoryContextAllocZero(xloginsert_cxt,
								   sizeof(registered_buffer) * (XLR_NORMAL_MAX_BLOCK_ID + 1));
		max_registered_buffers = XLR_NORMAL_MAX_BLOCK_ID + 1;
	}
	if (rdatas == NULL)
	{
		rdatas = MemoryContextAlloc(xloginsert_cxt,
									sizeof(XLogRecData) * XLR_NORMAL_RDATAS);
		max_rdatas = XLR_NORMAL_RDATAS;
	}

	/*
	 * 分配一个缓冲区以保存WAL记录的头部信息。
	 */
	if (hdr_scratch == NULL)
		hdr_scratch = MemoryContextAllocZero(xloginsert_cxt,
											 HEADER_SCRATCH_SIZE);
}
